package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;

/**
 * Author: Felix
 * Date: 2022/6/11
 * Desc: 将流中的数据写到phoenix表中
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private DruidDataSource dataSource;
    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtil.createDataSource();
    }

    //{"tm_name":"小米","sink_table":"dim_base_trademark","id":5}
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取phoenix维度表名称
        String tableName = jsonObj.getString("sink_table");
        //{"tm_name":"小米","id":5}
        jsonObj.remove("sink_table");

        //获取操作类型
        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        //拼接upsert语句
        String upsertSql = "upsert into "+ GmallConfig.PHOENIX_SCHEMA+"." + tableName +
            " ("+StringUtils.join(jsonObj.keySet(),",")+") values('"+StringUtils.join(jsonObj.values(),"','")+"')";

        System.out.println("向phoenix表中插入数据的sql:" + upsertSql);
        //通过连接池获取连接对象
        Connection conn = dataSource.getConnection();
        //执行upsert语句
        PhoenixUtil.executeSql(upsertSql,conn);

        //如果对维度进行了修改，删除Redis中缓存的维度
        if("update".equals(type)){
            DimUtil.deleteCached(tableName,jsonObj.getString("id"));
        }
    }
}
